import { SpanKind } from "@opentelemetry/api";
import { getLangWatchTracer } from "langwatch";
import type { SemConvAttributes } from "langwatch/observability";
import { createLogger } from "../../../../utils/logger";
import type {
  EventSourcedQueueDefinition,
  EventSourcedQueueProcessor,
} from "../../library/queues";

interface QueuedJob<Payload> {
  payload: Payload;
  jobId?: string;
  resolve: () => void;
  reject: (error: Error) => void;
}

/**
 * Minimal in-memory queue processor for dev/test environments.
 * Processes jobs asynchronously with simple concurrency control.
 *
 * **Use Cases:**
 * - Local development (when Redis is not available)
 * - Unit/integration tests
 * - Single-instance deployments
 *
 * **Limitations:**
 * - Not thread-safe (single process only)
 * - No persistence (jobs lost on restart)
 * - Simple concurrency (no advanced scheduling)
 */
export class EventSourcedQueueProcessorMemory<Payload>
  implements EventSourcedQueueProcessor<Payload>
{
  private readonly logger = createLogger("langwatch:event-sourcing:queue");
  private readonly tracer: ReturnType<typeof getLangWatchTracer>;
  private readonly queueName: string;
  private readonly process: (payload: Payload) => Promise<void>;
  private readonly spanAttributes?: (payload: Payload) => SemConvAttributes;
  private readonly makeJobId?: (payload: Payload) => string;
  private readonly delay?: number;
  private readonly concurrency: number;

  // Simple queue state
  private readonly queue: QueuedJob<Payload>[] = [];
  private readonly pendingJobs = new Map<string, QueuedJob<Payload>>();
  private activeCount = 0;
  private shutdownRequested = false;

  constructor(definition: EventSourcedQueueDefinition<Payload>) {
    const { name, process, spanAttributes, makeJobId, delay, options } =
      definition;

    this.tracer = getLangWatchTracer("langwatch.event-sourcing.queue");
    this.spanAttributes = spanAttributes;
    this.makeJobId = makeJobId;
    this.delay = delay;
    this.concurrency = options?.concurrency ?? 5;
    this.queueName = name;
    this.process = process;

    this.logger.info(
      { queueName: this.queueName, concurrency: this.concurrency },
      "Event-sourced queue processor initialized in memory mode (no Redis)",
    );
  }

  async send(payload: Payload): Promise<void> {
    // Memory implementation allows sends after close since it has no persistent state
    // This is different from BullMQ which should reject sends after shutdown

    const jobId = this.makeJobId ? this.makeJobId(payload) : void 0;

    // Simple job deduplication: replace existing job with same ID
    if (jobId) {
      const existingJob = this.pendingJobs.get(jobId);
      if (existingJob) {
        existingJob.payload = payload;
        this.logger.debug(
          { queueName: this.queueName, jobId },
          "Replaced existing job with same jobId",
        );
        return;
      }
    }

    // Queue job and process asynchronously
    return new Promise<void>((resolve, reject) => {
      const job: QueuedJob<Payload> = {
        payload,
        jobId,
        resolve,
        reject,
      };

      if (jobId) {
        this.pendingJobs.set(jobId, job);
      }

      this.queue.push(job);
      // Start processing if we have capacity
      this.tryProcessNext();
    });
  }

  /**
   * Processes next job from queue if capacity available.
   */
  private tryProcessNext(): void {
    // No capacity or no jobs
    // Memory implementation allows processing after shutdown since it has no persistent state
    if (this.activeCount >= this.concurrency || this.queue.length === 0) {
      return;
    }

    const job = this.queue.shift();
    if (!job) {
      return;
    }

    this.activeCount++;
    void this.processJob(job).finally(() => {
      this.activeCount--;
      if (job.jobId) {
        this.pendingJobs.delete(job.jobId);
      }
      // Try to process next job
      this.tryProcessNext();
    });
  }

  /**
   * Processes a single job with tracing and error handling.
   */
  private async processJob(job: QueuedJob<Payload>): Promise<void> {
    // Apply delay if configured
    if (this.delay && this.delay > 0) {
      await new Promise((resolve) => setTimeout(resolve, this.delay));
    }

    const baseAttributes: Record<string, string | number | boolean> = {
      "queue.name": this.queueName,
      "queue.job_id": job.jobId ?? "unknown",
    };

    let customAttributes: Record<string, string | number | boolean> = {};
    if (this.spanAttributes) {
      try {
        const attributes = this.spanAttributes(job.payload);
        // Filter out undefined values and convert to the expected type
        for (const [key, value] of Object.entries(attributes)) {
          if (value !== undefined) {
            if (
              typeof value === "string" ||
              typeof value === "number" ||
              typeof value === "boolean"
            ) {
              customAttributes[key] = value;
            }
          }
        }
      } catch (error) {
        // If spanAttributes throws, log error and continue with base attributes only
        const errorMessage =
          error instanceof Error ? error.message : String(error);
        this.logger.error(
          {
            queueName: this.queueName,
            jobId: job.jobId,
            error: errorMessage,
          },
          "Failed to extract span attributes from payload",
        );
      }
    }
    const attributes = { ...baseAttributes, ...customAttributes };

    try {
      await this.tracer.withActiveSpan(
        "pipeline.process",
        {
          kind: SpanKind.INTERNAL,
          attributes,
        },
        async () => {
          await this.process(job.payload);
        },
      );
      job.resolve();
    } catch (error) {
      const errorMessage =
        error instanceof Error ? error.message : String(error);
      this.logger.error(
        {
          queueName: this.queueName,
          jobId: job.jobId,
          error: errorMessage,
        },
        "Event-sourced queue job failed",
      );
      job.reject(error instanceof Error ? error : new Error(errorMessage));
    }
  }

  /**
   * Gracefully closes the queue processor, waiting for in-flight jobs to complete.
   */
  async close(): Promise<void> {
    this.logger.info(
      { queueName: this.queueName },
      "Closing memory queue processor",
    );

    this.shutdownRequested = true;

    // Wait for active jobs to complete (simple polling since we don't track promises)
    while (this.activeCount > 0) {
      await new Promise((resolve) => setTimeout(resolve, 10));
    }

    // Reject any remaining queued jobs
    for (const job of this.queue) {
      job.reject(
        new Error(
          `Queue ${this.queueName} was closed before job could be processed`,
        ),
      );
    }
    this.queue.length = 0;
    this.pendingJobs.clear();

    this.logger.info(
      { queueName: this.queueName },
      "Memory queue processor closed successfully",
    );
  }
}
